package cn.com.kingyea.es;

import java.io.File;
import java.io.FileInputStream;
import java.net.InetAddress;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Map.Entry;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;


public class ReIndexMain {

	private static Properties pro;
//	private String mypath = "D:/pro/env/ws/03/EsReindex/src/main/resources/es.conf";
	TransportClient esClient = null;
	BulkProcessor bulkProcessor = null;

	String toIndex = null;
	String toType = null;
	String pk = null;
	
	static String exchangeFields = null;

//	private static final Logger LOG = LoggerFactory.getLogger(ReIndexMain.class);
//	private static final Logger LOG = LogManager.getLogger(ReIndexMain.class);
	
	public static void main(String[] args) {
//		LOG.info("path: "+args[0]);
		System.out.println("path: "+args[0]);
		new ReIndexMain().execute(args[0]);
	}

	private static boolean addException = false;
	private void execute(String path) {
		
		try {
			System.out.println("start");
//			LOG.info("start");
			setPro(path);
			initParams();
			prepareClient();
			prepareProcessor();
			
			String[] fields = getPro().getProperty("from.index").split(";");
			String[] splitfields = getPro().getProperty("exchangeFields").split(";");//exchangeFields
			for(int i=0;i<fields.length && null !=fields[i];i++){
				exchangeFields = splitfields[i];
				String fromIndx = fields[i];
				if(fromIndx.endsWith("exception")){//把exception字段boolean添加。
					addException = true;
				}else {
					addException = false;
				}
				//TODO
				String[] includeFields = getIncludeFields(exchangeFields);
				// action
				SearchResponse searchResponse = esClient.prepareSearch(fromIndx)
						.setTypes(getPro().getProperty("from.type"))
						.setQuery(QueryBuilders.matchAllQuery())
						.setFetchSource(includeFields, null)
						.addSort("timestamp", SortOrder.ASC)
//						.addSort(SortBuilders.fieldSort("timestamp"))
						.setSize(30)
						// 这个游标维持多长时间
						.setScroll(TimeValue.timeValueMinutes(8)).execute().actionGet();

				for (SearchHit hit : searchResponse.getHits()) {
					Map<String, Object> source = hit.getSource();
					HashMap<String, Object> hashMap = new HashMap<>();
					// hashMap.put("_id", source.get("batchID"));
					if (source.containsKey(pk)) {
						exchangeData(getFrom2To(), source, hashMap);
						bulkProcessor.add(new UpdateRequest(toIndex, toType, source.get(pk).toString()).doc(hashMap).upsert(new IndexRequest(toIndex, toType, source.get(pk).toString()).source(hashMap)));
					}
				}
				bulkProcessor.flush();

//				System.out.println("------------------------------");
				// 使用上次的scrollId继续访问
//				ReIndexMain main = new ReIndexMain();
				do {
					bulkProcessor.flush();
					int num = scanData(esClient, bulkProcessor, searchResponse.getScrollId());
					if (num == 0)
						break;
				} while (true);
				System.out.println(fromIndx + "------------------------------END");
//				LOG.info(fromIndx + "------------------------------END");
			}
			
		} catch (Exception e) {
			e.printStackTrace();
//			LOG.error(e.getMessage(),e);
		}
	}

	private String[] getIncludeFields(String exchangeFields){
		String[] split = exchangeFields.split(",");
		ArrayList<String> arrayList = new ArrayList<String>();
		for(String s: split){
			if("exception".equals(s.split(":")[0])){
				continue;
			}
			arrayList.add(s.split(":")[0]);
		}
		String[] array = arrayList.toArray(new String[]{});
		return array;
	}
	private int scanData(Client esClient, BulkProcessor bulkProcessor, String scrollId) {

		SearchResponse searchResponse = esClient.prepareSearchScroll(scrollId).setScroll(TimeValue.timeValueMinutes(8))
				.execute().actionGet();
		int num = searchResponse.getHits().hits().length;
		for (SearchHit hit : searchResponse.getHits()) {
			Map<String, Object> source = hit.getSource();
			HashMap<String, Object> hashMap = new HashMap<>();

			if (source.containsKey(pk)) {
				exchangeData(getFrom2To(), source, hashMap);
				//new UpdateRequest("new2hshe", "hshe12","key2").doc(hashMap).upsert(new IndexRequest("new2hshe", "hshe12","key2").source(hashMap))
				bulkProcessor.add(new UpdateRequest(toIndex, toType, source.get(pk).toString()).doc(hashMap).upsert(new IndexRequest(toIndex, toType, source.get(pk).toString()).source(hashMap)));
			}
		}
		return num;
	}

	private void initParams() {
		toIndex = getPro().getProperty("to.index");
		toType = getPro().getProperty("to.type");
		pk = getPro().getProperty("es.pk");
	}

	private void prepareProcessor() {
		bulkProcessor = BulkProcessor.builder(esClient, new BulkProcessor.Listener() {
			@Override
			public void beforeBulk(long executionId, BulkRequest request) {
//				System.out.println("before");
			}

			@Override
			public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
//				System.out.println("afterbulk");
			}

			@Override
			public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
//				System.out.println("after");
			}
		}).setBulkActions(10000).setBulkSize(new ByteSizeValue(50, ByteSizeUnit.MB))
				.setFlushInterval(TimeValue.timeValueSeconds(5)).setConcurrentRequests(1)
				.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(1000), 3)).build();

	}

	private void prepareClient() throws Exception {
		String clusterName = getPro().getProperty("cluster.name");
		String[] ips = getPro().getProperty("hosts").split(",");

		Settings settings = Settings.builder().put("cluster.name", clusterName).build();
		// 创建client
		esClient = new PreBuiltTransportClient(settings);
		for (int i = 0; i < ips.length && ips[i].length() > 1; i++) {
			esClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ips[i]), 9300));
		}

	}

	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
	private static void exchangeData(Map<String, String> fromTo, Map<String, Object> fromMap,
			Map<String, Object> toMap) {
		if (addException) {//存在exception字段。
			toMap.put("exception", true);
		}
		for (Entry<String, String> e : fromTo.entrySet()) {
			String key = e.getKey();
			if (fromMap.containsKey(key)) {
				if("endTime".equals(key) || "beginTime".equals(key)) {//特殊。
					toMap.put("@"+e.getValue(), sdf.format(new Date(Long.valueOf(fromMap.get(key).toString()))));
					toMap.put(e.getValue(), fromMap.get(key));
				}else {
					toMap.put(e.getValue(), fromMap.get(key));
				}
			}
		}
	}

	public static Properties getPro() {
		return pro;
	}

	public void setPro(String path) throws Exception {
		Properties properties = new Properties();
		properties.load(new FileInputStream(new File(path)));
		this.pro = properties;
	}

	private static Map<String, String> getFrom2To() {
		try {
			HashMap<String, String> fromTo = new HashMap<String, String>();
			Properties properties = getPro();
//			String str = properties.getProperty("exchangeFields");
			
			String[] split = exchangeFields.split(",");
			for (String s : split) {
				String[] split2 = s.split(":");
				fromTo.put(split2[0], split2[1]);
			}
			return fromTo;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
}
